package com.zt.bigdata.elasticsearch.utils.transport.apis;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.zt.bigdata.elasticsearch.utils.transport.ESClient;
import org.apache.commons.collections.CollectionUtils;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.*;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.script.Script;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;


/**
 * @ClassName DocApi
 * @Description
 * @Author zhangtonghy
 * @Date 2019-02-21 14:52
 * @Copyright: 版权所有 (C) zt zt.
 * @注意：本内容仅限于zt内部传阅，禁止外泄以及用于其他的商业目的
 **/
public class DocApi {

    private TransportClient transportClient;
    private ObjectMapper mapper = new ObjectMapper();
    private static final Logger logger = LoggerFactory.getLogger(DocApi.class);

    public TransportClient getClient() {
        if (null == transportClient) {
            ESClient esClient = new ESClient(" application-dev", "172.28.23.35:9300");
            transportClient = esClient.getClient();
        }
        return transportClient;
    }

    public RestStatus indexByMap(String index, String type, String id, Map<String, Object> map) {
        return getClient().prepareIndex(index, type, id)
                .setSource(map).get().status();
    }

    public RestStatus indexByMap(String index, String type, Map<String, Object> map) {
        return indexByMap(index, type, null, map);
    }

    public RestStatus indexByJsonString(String index, String type, String jsonString) {
        return indexByJsonString(index, type, null, jsonString);
    }

    public RestStatus indexByJsonString(String index, String type, String id, String jsonString) {
        return getClient().prepareIndex(index, type, id)
                .setSource(jsonString, XContentType.JSON)
                .get().status();
    }

    public GetResponse getDocById(String index, String type, String id) {
        return getClient().prepareGet(index, type, id).get();
    }

    public RestStatus deleteById(String index, String type, String id) {
        return getClient().prepareDelete(index, type, id).get().status();

    }

    public BulkByScrollResponse synDeleteByQuery(String index, String field, String... values) {
        return synDeleteByQuery(index, QueryBuilders.multiMatchQuery(field, values));
    }

    public BulkByScrollResponse synDeleteByQuery(String index, QueryBuilder builder) {
        return DeleteByQueryAction.INSTANCE.newRequestBuilder(getClient())
                .filter(builder)
                .source(index)
                .get();
    }

    public void asynDeleteByQuery(String index, String field, String... values) {
        asynDeleteByQuery(index, QueryBuilders.multiMatchQuery(field, values));
    }

    public void asynDeleteByQuery(String index, QueryBuilder builder) {
        DeleteByQueryAction.INSTANCE.newRequestBuilder(getClient())
                .filter(builder)
                .source(index)
                .execute(new ActionListener<BulkByScrollResponse>() {
                    @Override
                    public void onResponse(BulkByScrollResponse response) {
                        response.getDeleted();
                    }

                    @Override
                    public void onFailure(Exception e) {
                        logger.error("e:{},message:{}", e.getCause(), e.getMessage());
                    }
                });
    }


    public RestStatus updateById(String index, String type, String id, Script script) throws ExecutionException, InterruptedException {
        return getClient()
                .prepareUpdate(index, type, id)
                .setScript(script)
                .get()
                .status();
    }

    /**
     * 如果 更新字段不存在 则插入
     *
     * @param index
     * @param type
     * @param id
     * @param map
     * @return
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public RestStatus updateAndMergeById(String index, String type, String id, Map<String, Object> map) throws ExecutionException, InterruptedException {
        UpdateRequest updateRequest = generateUpdateRequest(index, type, id).doc(map);
        return getClient().update(updateRequest).get().status();
    }

    public RestStatus upsertById(String index, String type, String id, Map<String, Object> updateFields, IndexRequest indexRequest) throws ExecutionException, InterruptedException {
        UpdateRequest updateRequest = generateUpdateRequest(index, type, id)
                .doc(updateFields)
                .upsert(indexRequest);
        return getClient().update(updateRequest).get().status();
    }


    public UpdateRequest generateUpdateRequest(String index, String type, String id) {
        UpdateRequest updateRequest = new UpdateRequest();
        updateRequest.index(index);
        updateRequest.type(type);
        updateRequest.id(id);
        return updateRequest;
    }

    public BulkItemResponse[] bulkRequest(List<IndexRequestBuilder> indexRequestBuilders) {
        if (CollectionUtils.isNotEmpty(indexRequestBuilders)) {
            BulkRequestBuilder bulkRequest = getClient().prepareBulk();
            indexRequestBuilders.forEach(builder -> bulkRequest.add(builder));
            BulkResponse bulkResponse = bulkRequest.get();
            if (bulkResponse.hasFailures()) {
                return bulkResponse.getItems();
            }
        }
        return null;
    }

    public static void main(String[] args) throws IOException {

        DocApi docApi = new DocApi();


//        JsonNode jsonNode = $.json.toJson(Files.readAllLines(Paths.get("/Users/zhangtong/Desktop/a.txt"))
//                .stream()
//                .collect(
//                        StringBuilder::new,
//                        (x, b) -> x.append(b),
//                        (x, b) -> x.append(b)).toString()
//        );
//        JsonNode hits = jsonNode.path("hits").path("hits");
//        List<IndexRequestBuilder> list = new ArrayList<>();
//        if (hits instanceof ArrayNode) {
//            ArrayNode arrayNode = ((ArrayNode) hits);
//            for (int i = 0; i < arrayNode.size(); i++) {
//                JsonNode _source = arrayNode.get(i).path("_source");
//                list.add(docApi.getClient().prepareIndex("hy-bigdata", "log").setSource(_source.toString(), XContentType.JSON));
//            }
//        }
//        BulkItemResponse[] responses = docApi.bulkRequest(list);
//        if (null != responses) {
//            Arrays.asList(responses).forEach(r -> System.out.println(r.getFailureMessage()));
//        }

        //39.134.86.164

        ESClient esClient = new ESClient("cdn-bigdata","39.134.86.164:9300");

    }

    /**
     * updateByQuery的最简单用法是更新索引中的每个文档而不更改源。 此用法可以启用新属性或其他在线映射更改。
     *
     * @param index
     * @return
     */
    public BulkByScrollResponse updateByQuery(String index) {
        return UpdateByQueryAction
                .INSTANCE
                .newRequestBuilder(getClient())
                .source(index)
                .abortOnVersionConflict(false)
                .get();
    }

    public BulkByScrollResponse updateByQuery(String index, QueryBuilder queryBuilder, Script script) {
        UpdateByQueryRequestBuilder updateByQuery =
                UpdateByQueryAction.INSTANCE.newRequestBuilder(getClient());
        updateByQuery.source(index)
                .filter(queryBuilder)
                .size(1000)
                .script(script);
        return updateByQuery.get();
    }


    public BulkByScrollResponse reIndex(String sourceIndex, String targetIndex, QueryBuilder queryBuilder) {
        return ReindexAction.INSTANCE.newRequestBuilder(getClient())
                .source(sourceIndex)
                .destination(targetIndex)
                .filter(queryBuilder)
                .get();
    }


}
